// Package broadcast broadcast protocol
package broadcast

import (
	"context"
	"encoding/hex"
	"sync"
	"time"

	"github.com/golang/protobuf/proto"
	lru "github.com/hashicorp/golang-lru"
	logger "hundsun.com/hsl/hschain/common/log"
	comps "hundsun.com/hsl/hschain/common/pubsub"
	"hundsun.com/hsl/hschain/common/types"
	"hundsun.com/hsl/hschain/common/util/pb"
	"hundsun.com/hsl/hschain/p2p/common"
	"hundsun.com/hsl/hschain/p2p/common/constant"
	"hundsun.com/hsl/hschain/p2p/impl/dht/protocol"
	ltypes "hundsun.com/hsl/hschain/p2p/impl/dht/protocol/broadcast/types"
	pbledger "hundsun.com/hsl/hschain/protos/ledger"
	"hundsun.com/hsl/hschain/store/mq"
)

var once sync.Once

var protocolInstance *Protocol

func init() {
	protocol.RegisterProtocolIniter(InitProtocol)
}

//Protocol Protocol
type Protocol struct {
	*protocol.P2PEnv
	Comps       *comps.PubSub
	TxFilter    *lru.Cache
	BlockFilter *lru.Cache
	txMonitor   *TxMonitor
}

//TxMonitor ...
type TxMonitor struct {
	txArr         []*pbledger.Transaction
	txAddChan     chan *pbledger.Transaction
	broadCstTxBuf chan *pbledger.Transaction
}

//InitTxMonitor monitor初始化
func InitTxMonitor() *TxMonitor {
	instanceTxPendingInstance := &TxMonitor{
		txAddChan:     make(chan *pbledger.Transaction, constant.TxaddchanSize),
		broadCstTxBuf: make(chan *pbledger.Transaction, constant.BroadCstTxBufSize),
	}
	return instanceTxPendingInstance
}

// InitProtocol init protocol
func InitProtocol(env *protocol.P2PEnv) {
	protocolInstance = &Protocol{}
	protocolInstance.Init(env)
}

//Init 初始化
func (p *Protocol) Init(env *protocol.P2PEnv) {
	p.P2PEnv = env
	p.txMonitor = InitTxMonitor()

	p.Comps = comps.NewPubSub(ltypes.EventPubsubCapacity)
	p.TxFilter, _ = lru.New(ltypes.TxRecvFilterCacheNum)
	p.BlockFilter, _ = lru.New(ltypes.BlockRecvFilterCacheNum)

	//注册事件处理函数
	once.Do(func() {
		protocol.RegisterEventHandler(types.EventTxBroadcast, p.handleTxBroadCastEvent)
		protocol.RegisterEventHandler(types.EventBlockBroadcast, p.handleBlockBroadCastEvent)
	})

	go p.TxPendingMonitor(p.Ctx, p.txMonitor.txAddChan)

	if p.Cfg.BroadcastType == ltypes.BroadcastV0 {
		// v0版本pubsub broadcast
		go NewBroadcastV0(p).Init()
	} else if p.Cfg.BroadcastType == ltypes.BroadcastV1 {
		// v1版本广播
		go NewBroadcastV1(p).Init()
	}
}

// 处理系统广播发送事件，交易
func (p *Protocol) handleTxBroadCastEvent(msg *mq.Message) {
	if msg == nil || msg.Data == nil {
		return
	}
	if tx, ok := msg.Data.(*pbledger.Transaction); ok {
		p.txMonitor.txAddChan <- tx
	}
}

//PushTx 发送交易
func (p *Protocol) PushTx() {
	if len(p.txMonitor.txArr) == 0 {
		return
	}

	p.txBroadCast(p.txMonitor.txArr)
	err := p.P2PEnv.TxPool.AddTransactions(p.txMonitor.txArr)
	if err != nil {
		logger.Errorf("push batch tx failed, reason: %v", err)
		return
	}
	logger.Tracef("push batch tx success, num:%d", len(p.txMonitor.txArr))
	p.txMonitor.txArr = []*pbledger.Transaction{}
}

//TxPendingMonitor 交易监控
func (p *Protocol) TxPendingMonitor(ctx context.Context, addChan <-chan *pbledger.Transaction) {
	dura := time.Millisecond * constant.BroadCstTxMaxWaitTime
	batchCount := constant.BroadCstTxBatchCount
	cfg, err := common.GetConfig()
	if err == nil {
		batchCount = cfg.BroadCastTxBatchCount
		dura = time.Millisecond * time.Duration(cfg.BroadCastTxMaxWaitTime)
	}
	ticker := time.NewTicker(dura)
	for {
		select {
		case <-ctx.Done():
			logger.Infof("tx pending monitor closed")
			return
		case tx := <-addChan:
			p.txMonitor.txArr = append(p.txMonitor.txArr, tx)
			if len(p.txMonitor.txArr) >= batchCount {
				p.PushTx()
			}
		case <-ticker.C:
			p.PushTx()
		}
	}
}

func (p *Protocol) txBroadCast(txs []*pbledger.Transaction) {
	if len(txs) == 0 {
		return
	}
	var filterTxs []*pbledger.Transaction
	for _, tx := range txs {
		hash := hex.EncodeToString(tx.Hash())
		if !p.TxFilter.Contains(hash) {
			p.TxFilter.Add(hash, struct{}{})
			filterTxs = append(filterTxs, tx)
		}
	}
	if len(filterTxs) == 0 {
		return
	}
	toBroadcast := &pbledger.Transactions{
		Txs: filterTxs,
	}
	//封装pb.Message
	buf, err := proto.Marshal(toBroadcast)
	if err != nil {
		p.Log.Errorf("txs serialize failed, reason: %v", err)
		return
	}
	msg := p.P2pNode.NewP2pMessage("", "", constant.MsgBroadcastTransaction, buf, nil)
	p.Comps.FIFOPub(msg, ltypes.TxTopic)
	p.Log.Infof("batch broadcast txs, num:%d", len(txs))
}

// 处理系统广播发送事件，区块
func (p *Protocol) handleBlockBroadCastEvent(msg *mq.Message) {
	p.Log.Debug("broadcast block request")
	if msg == nil || msg.Data == nil {
		return
	}
	if block, ok := msg.Data.(*pbledger.Block); ok {
		hash := hex.EncodeToString(block.Hash())
		if !p.BlockFilter.Contains(hash) {
			p.BlockFilter.Add(hash, struct{}{})
			buf, err := proto.Marshal(block)
			if err != nil {
				p.Log.Errorf("block serialize failed, reason: %v", err)
				return
			}
			if len(buf) > int(pb.GetGenesisTransactionPayload().BlockLimit.MaxBlockSizeBytes) {
				p.Log.Errorf("block size exceed limit")
				return
			}
			msg := p.P2pNode.NewP2pMessage("", "", constant.MsgBroadcastBlock, buf, nil)
			p.Comps.FIFOPub(msg, ltypes.BlockTopic)
			p.Log.Infof("broadcast block, blkNumber:%d", block.Header.Number)
		}
	}
}
